package eventBus0421

import (
	"go.uber.org/zap"
	"runtime"
	"sync"
	"time"
)

type TaskFunc func()

type asyncTask struct {
	fn TaskFunc
}

type subscriber struct {
	handler func(any)
}

type Bus struct {
	subNode     sync.Map
	taskQueue   chan asyncTask
	shutdownSig chan struct{}
	closed      bool
	mu          sync.Mutex

	tpsCounter   *TPSCounter
	metrics      *Metrics
	logger       *zap.Logger
	workerMng    *workerManager
	workerCount  int32
	maxQueueSize int
}

func NewBus(queueSize int, logger *zap.Logger) *Bus {
	cpuNum := runtime.NumCPU()
	b := &Bus{
		taskQueue:    make(chan asyncTask, queueSize),
		shutdownSig:  make(chan struct{}),
		tpsCounter:   NewTPSCounter(10),
		metrics:      &Metrics{},
		logger:       logger,
		maxQueueSize: queueSize,
	}
	b.workerMng = newWorkerManager(cpuNum, func() { b.spawnWorker() })
	for i := 0; i < cpuNum; i++ {
		b.spawnWorker()
	}
	// go b.monitor() // 启动监控协程
	return b
}

// 发布消息
func (b *Bus) Publish(topic string, msg any) {
	b.mu.Lock()
	if b.closed {
		b.mu.Unlock()
		return
	}
	b.mu.Unlock()

	b.tpsCounter.Record()
	value, ok := b.subNode.Load(topic)
	if !ok {
		b.metrics.IncDrop()
		return
	}
	sub := value.(*subscriber)

	task := asyncTask{
		fn: func() {
			defer func() {
				if err := recover(); err != nil {
					b.logger.Warn("panic in subscriber", zap.Any("err", err))
				}
			}()
			sub.handler(msg)
		},
	}

	select {
	case b.taskQueue <- task:
		b.metrics.IncPublished()
	default:
		b.metrics.IncDrop()
	}
}

// 订阅主题
func (b *Bus) Subscribe(topic string, handler func(any)) {
	b.subNode.Store(topic, &subscriber{handler: handler})
}

func (b *Bus) spawnWorker() {
	go func() {
		b.workerCount++ // 增加 worker 数量
		//b.logger.Info("Starting new worker", zap.Int32("currentWorkerCount", b.workerCount)) // 日志输出
		defer func() {
			b.workerCount-- // 退出时减少 worker 数量
			//b.logger.Info("Worker stopped", zap.Int32("currentWorkerCount", b.workerCount)) // 日志输出
		}()

		for {
			select {
			case task := <-b.taskQueue:
				task.fn()
			case <-b.shutdownSig:
				return
			}
		}
	}()
}


// 停止事件总线
func (b *Bus) Close() {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.closed {
		return
	}
	b.closed = true
	close(b.shutdownSig)
}

// 监控指标打印
func (b *Bus) monitor() {
	ticker := time.NewTicker(1 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			b.logger.Info("[Metrics]",
				zap.Int64("Published", b.metrics.Published()),
				zap.Int64("Dropped", b.metrics.Dropped()),
				zap.Int("QueueLength", len(b.taskQueue)),
				zap.Int64("TPS_1s", b.tpsCounter.TPS(1)),
				zap.Int64("TPS_5s", b.tpsCounter.TPS(5)),
				zap.Int64("TPS_10s", b.tpsCounter.TPS(10)),
			)
			b.workerMng.Evaluate(len(b.taskQueue))
		case <-b.shutdownSig:
			return
		}
	}
}
